#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <getopt.h>
#include <string.h>
#include <pthread.h>
#include <errno.h>

#include "perf.h"
#include "base.h"
#include "json/json.h"

TestConf *tf;
SafeQueue *sq;

void *worker(void *arg)
{
    int ret;
    while (1) {
        QueueItem *item = safequeue_get(sq);
        if (NULL == item && sq->is_over == 1) {
            //printf("%d thread is over\n", pthread_self());
            return NULL;
        }
        //printf("%d I am worker thread: %s\n", pthread_self(), item->json);
        //do sthing
        zz_free(item->json);
        zz_free(item);
    }
    return NULL;
}

void *producer(void *arg)
{
    int count = 0;
    int ret;
    char *test = NULL;
    char in = 'a';

    test = (char *)zz_malloc(tf->itemsize + 1);
    if (test == NULL) {
        DERROR("malloc error\n");
        return NULL;
    }
    memset(test, in, tf->itemsize);
    *(test + tf->itemsize) = '\0';

    //printf("%d I am producer thread\n", pthread_self());
    while (1) {
        int i;
        for (i = 0; i < tf->dcount_per_producer; i++) {
            Json::Value root;
            Json::Value arrayObj;
            Json::Value item;

            int k = 0;
            for (k = 0; k < tf->lines_per_data; k++) {
                TableItem *ptr = tf->head; 
                char buf[1024];
                int j = 0;
                while (ptr) {
                    if (j == 0) {
                        snprintf(buf, 1024, "%d_%010d", pthread_self(), count);
                        item[ptr->name] = buf;
                    } else {
                        //snprintf(buf, 1024, "%s_%d", "test", j);
                        item[ptr->name] = test;
                    }
                    j++;
                    ptr = ptr->next;
                }
                arrayObj.append(item);
                count++;
            }
            root["array"] = arrayObj;
            std::string jsonstr = root.toStyledString();
            //std::cout << jsonstr << std::endl;

            int len = strlen(jsonstr.c_str());
            char *json = (char *)zz_malloc(len + 1);
            memset(json, 0x0, len + 1);
            strcpy(json, jsonstr.c_str());

           safequeue_put(sq, json);
        }
        break;
    }
    zz_free(test);
    if (sq->dcount == tf->producer_thread_nums * tf->dcount_per_producer) {
        printf("data count: %d\n", sq->dcount);
        sq->is_over = 1;
        ret = pthread_cond_broadcast(&sq->cond_full);
        if (ret != 0) {
            char errbuf[1024];
            strerror_r(errno, errbuf, 1024);
            DERROR("pthread_cond_broadcast error: %s\n", errbuf);
            exit(-1);
        }
        //printf("------------------\n");
    }
    return NULL;
}

int 
start_thread_pool()
{
    int j = 0, i;
    int ret;
    int nums = tf->work_thread_nums + tf->producer_thread_nums;
    pthread_t threads[nums];
    
    //create producer thread
    for (i = 0; i < tf->producer_thread_nums; i++) {
        ret = pthread_create(&threads[j], NULL, producer, NULL);
        if (ret < 0) {
            char errbuf[1024];
            strerror_r(errno, errbuf, 1024);
            DERROR("pthread_create error! %s\n", errbuf);
            return -1;
        }
        j++;
    }

    //create worker thread
    for (i = 0; i < tf->work_thread_nums; i++) {
        ret = pthread_create(&threads[j], NULL, worker, NULL);
        if (ret < 0) {
            char errbuf[1024];
            strerror_r(errno, errbuf, 1024);
            DERROR("pthread_create error! %s\n", errbuf);
            return -1;
        }
        j++;
    }
    
    for (i = 0; i < nums; i++) {
        pthread_join(threads[i], NULL);
    }

    return 0;
}

void*
thread_func(void *arg)
{
    int ret;
    int count = 0;

    /*char *test = NULL;
    int  in = 'a';

    test = (char *)zz_malloc(tf->itemsize + 1);
    if (test == NULL) {
        DERROR("malloc error\n");
        return NULL;
    }
    memset(test, in, tf->itemsize);
    *(test + tf->itemsize) = '\0';
    
    int i;
    for (i = 0; i < tf->dcount_per_producer; i++) {
        Json::Value root;
        Json::Value arrayObj;
        Json::Value item;

        int k = 0;
        for (k = 0; k < tf->lines_per_data; k++) {
            TableItem *ptr = tf->head;
            char buf[1024];
            int j = 0;
            while (ptr) {
                if (j == 0) {
                    snprintf(buf, 1024, "%d_%010d", pthread_self(), count);
                    item[ptr->name] = buf;
                } else {
                    item[ptr->name] = test;
                }
                j++;
                ptr = ptr->next;
            }
            arrayObj.append(item);
            count++;
        }
        root["array"] = arrayObj;
        std::string jsonstr = root.toStyledString();
        pthread_mutex_lock(&sq->lock);
        sq->dcount++;
        pthread_mutex_unlock(&sq->lock);

    }
    zz_free(test);*/

    insert();
    printf("thread %u over\n", pthread_self());
    return NULL;
}

int
start_thread_multiple()
{
    int i, ret;
    int nums = tf->work_thread_nums;
    pthread_t threads[nums];
    
    for (i = 0; i < nums; i++) {
        ret = pthread_create(&threads[i], NULL, thread_func, NULL);
        if (ret < 0) {
            char errbuf[1024];
            strerror_r(errno, errbuf, 1024);
            DERROR("pthread_create error! %s\n", errbuf);
            return -1;
        }
    }

    for (i = 0; i < nums; i++) {
        pthread_join(threads[i], NULL);
    }
    printf("data count: %d\n", sq->dcount);
    return 0;
}

void
show_help()
{
    printf("usage:\n\tperf [option]\n");
    printf("options:\n");
    printf("\t--tablename, -t\ttable name\n");
    printf("\t--items, -i\ttable items(item1:item2:item3:item4)\n");
    printf("\t--worknum, -w\twork thread nums\n");
    printf("\t--producernum, -p\tproducer thread nums\n");
    printf("\t--count, -c\tdata count per producer\n");
    printf("\t--lines, -l\thow mutch lines in one data\n");
    printf("\t--size, -s\thow mutch size per item(byte)\n");
    printf("\t--qsize, -q\tsafe queue size\n");
    printf("\n\n");

    exit(0);
}

int main(int argc, char **argv)
{
    int ret;
    logfile_create("test.log", 4);
    
    int    optret;
    int    optidx = 0;
    char   *optstr = "t:i:w:p:c:l:s:q:";

    struct option myoption[] = {
        {"tablename", 0, NULL, 't'},
        {"items", 0, NULL, 'i'},
        {"worknum", 0, NULL, 'w'},
        {"producernum", 0, NULL, 'p'},
        {"count", 0, NULL, 'c'},
        {"lines", 0, NULL, 'l'},
        {"size", 0, NULL, 's'},
        {"qsize", 0, NULL, 'q'},
        {NULL, 0, NULL, 0}
    };

    if (argc < 2) {
        show_help();
        return -1;
    }
    
    tf = (TestConf *)zz_malloc(sizeof(TestConf));
    memset(tf, 0, sizeof(TestConf));

    tf->mode = SINGLE_THREAD;

    while(optret = getopt_long(argc, argv, optstr, myoption, &optidx)) {
        if (optret < 0) {
            break;
        }
        
        switch(optret) {
            case 't':
                snprintf(tf->tname, 255, "%s", optarg);
                break;
            case 'i':
            {
                char buffer[1024] = {0};
                snprintf(buffer, 1024, "%s", optarg);
                
                char *ptr = NULL;
                char *start = NULL;

                start = ptr = buffer;
                TableItem *tail = tf->head;
                while (ptr) {
                    ptr = strchr(ptr, ':');
                    if (ptr) {
                        *ptr = '\0';
                        ptr++;
                        TableItem *item = (TableItem *)zz_malloc(sizeof(TableItem));
                        if (item == NULL) {
                            DERROR("malloc tableitem struct error\n");
                            exit(-1);
                        }
                        strcpy(item->name, start);
                        //item->next = tf->head;
                        //tf->head = item;
                        if (!tail) {
                            tf->head = item;
                            item->next = NULL;
                            tail = tf->head;
                        } else {
                            tail->next = item;
                            item->next = NULL;
                            tail = item;
                        }
                        start = ptr;
                    } else {
                        TableItem *item = (TableItem *)zz_malloc(sizeof(TableItem));
                        if (item == NULL) {
                            DERROR("malloc table item struct error\n");
                            exit(-1);
                        }
                        strcpy(item->name, start);
                        //item->next = tf->head;
                        //tf->head = item;
                        if (!tail) {
                            tf->head = item;
                            item->next = NULL;
                            tail = tf->head;
                        } else {
                            tail->next = item;
                            item->next = NULL;
                            tail = item;
                        }
                    }
                }
                break;
            }
            case 'w':
                tf->work_thread_nums = atoi(optarg);
                break;
            case 'p':
                tf->producer_thread_nums = atoi(optarg);
                break;
            case 'c':
                tf->dcount_per_producer = atoi(optarg);
                break;
            case 'l':
                tf->lines_per_data = atoi(optarg);
                break;
            case 's':
                tf->itemsize = atoi(optarg);
                break;
            case 'q':
                tf->qsize = atoi(optarg);
                break;
            default:
                printf("error option: %c\n", optret);
                exit(-1);
                break;
        }
    }
    
    DINFO("perfomance opt t: %s w: %d p: %d d: %d s: %d\n", tf->tname, tf->work_thread_nums, tf->producer_thread_nums, tf->dcount_per_producer, tf->qsize);
    
    TableItem *item = tf->head;
    while (item) {
        DINFO("item: %s\n", item->name);
        item = item->next;
    }
    if (tf->producer_thread_nums > 0) {
        tf->mode = THREAD_POOL;
        printf("thead pool dispos\n");
        sq = safequeue_create();
        ret = start_thread_pool();
    } else if (tf->work_thread_nums > 1) {
        tf->mode = MULTIPLE_THREAD;
        printf("multiple thread dispose\n");
        sq = safequeue_create();
        ret = start_thread_multiple();
    } else {
        tf->mode = SINGLE_THREAD;
        thread_func(NULL);
        DINFO("single thread dispose\n");
    }
    return 0;
}
